<!DOCTYPE html>
<html>
<head>
  <meta charset="utf-8">
  
  <title>zookeeper源码分析-事件注册及通信原理 | Eetal&#39;s Blog</title>

  <!-- keywords -->
  

  <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
  <meta name="description" content="欢迎查看Eetal的第二十六篇博客–zookeeper源码分析-事件注册及通信原理 zookeeper基本命令代码依赖123&amp;lt;groupId&amp;gt;org.apache.zookeeper&amp;lt;/groupId&amp;gt;&amp;lt;artifactId&amp;gt;zookeeper&amp;lt;/artifactId&amp;gt;&amp;lt;version&amp;gt;3.5.5&amp;lt;/version&amp;gt; demo">
<meta property="og:type" content="article">
<meta property="og:title" content="zookeeper源码分析-事件注册及通信原理">
<meta property="og:url" content="http://eetal.gitee.io/blog/2019/07/31/zookeeper源码分析-事件注册及通信原理/index.html">
<meta property="og:site_name" content="Eetal&#39;s Blog">
<meta property="og:description" content="欢迎查看Eetal的第二十六篇博客–zookeeper源码分析-事件注册及通信原理 zookeeper基本命令代码依赖123&amp;lt;groupId&amp;gt;org.apache.zookeeper&amp;lt;/groupId&amp;gt;&amp;lt;artifactId&amp;gt;zookeeper&amp;lt;/artifactId&amp;gt;&amp;lt;version&amp;gt;3.5.5&amp;lt;/version&amp;gt; demo">
<meta property="og:locale" content="default">
<meta property="og:updated_time" content="2019-07-31T18:18:19.409Z">
<meta name="twitter:card" content="summary">
<meta name="twitter:title" content="zookeeper源码分析-事件注册及通信原理">
<meta name="twitter:description" content="欢迎查看Eetal的第二十六篇博客–zookeeper源码分析-事件注册及通信原理 zookeeper基本命令代码依赖123&amp;lt;groupId&amp;gt;org.apache.zookeeper&amp;lt;/groupId&amp;gt;&amp;lt;artifactId&amp;gt;zookeeper&amp;lt;/artifactId&amp;gt;&amp;lt;version&amp;gt;3.5.5&amp;lt;/version&amp;gt; demo">
  
    <link rel="alternative" href="/atom.xml" title="Eetal&#39;s Blog" type="application/atom+xml">
  
  
    <link rel="icon" href="http://www.yangyitao.top/FiveChess/resource/ico/favicon.ico">
  
  <link rel="stylesheet" href="/blog/css/style.css">
  
  

  <script src="//cdn.bootcss.com/require.js/2.3.2/require.min.js"></script>
  <script src="//cdn.bootcss.com/jquery/3.1.1/jquery.min.js"></script>

  
</head>
<body>
  <div id="container">
    <div id="particles-js"></div>
    <div class="left-col">
    <div class="overlay"></div>
<div class="intrude-less">
	<header id="header" class="inner">
		<a href="/" class="profilepic">
			
			<img lazy-src="/blog/images/head.jpg" class="js-avatar">
			
		</a>

		<hgroup>
		  <h1 class="header-author"><a href="/">Eetal</a></h1>
		</hgroup>

		
		<p class="header-subtitle">java coder</p>
		

		
			<div class="switch-btn">
				<div class="icon">
					<div class="icon-ctn">
						<div class="icon-wrap icon-house" data-idx="0">
							<div class="birdhouse"></div>
							<div class="birdhouse_holes"></div>
						</div>
						<div class="icon-wrap icon-ribbon hide" data-idx="1">
							<div class="ribbon"></div>
						</div>
						
						<div class="icon-wrap icon-link hide" data-idx="2">
							<div class="loopback_l"></div>
							<div class="loopback_r"></div>
						</div>
						
						
					</div>
					
				</div>
				<div class="tips-box hide">
					<div class="tips-arrow"></div>
					<ul class="tips-inner">
						<li>菜单</li>
						<li>标签</li>
						
						<li>友情链接</li>
						
						
					</ul>
				</div>
			</div>
		

		<div class="switch-area">
			<div class="switch-wrap">
				<section class="switch-part switch-part1">
					<nav class="header-menu">
						<ul>
						
							<li><a href="/blog/">Home</a></li>
				        
							<li><a href="/blog/archives">Archives</a></li>
				        
						</ul>
					</nav>
					<nav class="header-nav">
						<div class="social">
							
						</div>
					</nav>
				</section>
				
				
				<section class="switch-part switch-part2">
					<div class="widget tagcloud" id="js-tagcloud">
						
					</div>
				</section>
				
				
				
				<section class="switch-part switch-part3">
					<div id="js-friends">
					
			          <a target="_blank" class="main-nav-link switch-friends-link" href="https://github.com/smackgg/hexo-theme-smackdown">smackdown</a>
			        
			        </div>
				</section>
				

				
			</div>
		</div>
	</header>				
</div>
    </div>
    <div class="mid-col">
      <nav id="mobile-nav">
  	<div class="overlay">
  		<div class="slider-trigger"></div>
  		<h1 class="header-author js-mobile-header hide">Eetal</h1>
  	</div>
	<div class="intrude-less">
		<header id="header" class="inner">
			<div class="profilepic">
				<img lazy-src="/blog/images/head.jpg" class="js-avatar">
			</div>
			<hgroup>
			  <h1 class="header-author">Eetal</h1>
			</hgroup>
			
			<p class="header-subtitle">java coder</p>
			
			<nav class="header-menu">
				<ul>
				
					<li><a href="/blog/">Home</a></li>
		        
					<li><a href="/blog/archives">Archives</a></li>
		        
		        <div class="clearfix"></div>
				</ul>
			</nav>
			<nav class="header-nav">
				<div class="social">
					
				</div>
			</nav>
		</header>				
	</div>
</nav>
      <div class="body-wrap"><article id="post-zookeeper源码分析-事件注册及通信原理" class="article article-type-post" itemscope itemprop="blogPost">
  
    <div class="article-meta">
      <a href="/blog/2019/07/31/zookeeper源码分析-事件注册及通信原理/" class="article-date">
  	<time datetime="2019-07-31T12:24:19.197Z" itemprop="datePublished">2019-07-31</time>
</a>
    </div>
  
  <div class="article-inner">
    
      <input type="hidden" class="isFancy" />
    
    
      <header class="article-header">
        
  
    <h1 class="article-title" itemprop="name">
      zookeeper源码分析-事件注册及通信原理
      
    </h1>
  

      </header>
      
      <div class="article-info article-info-post">
        
        

        
        <div class="clearfix"></div>
      </div>
      
    
    <div class="article-entry" itemprop="articleBody">
      
        <p>欢迎查看Eetal的第二十六篇博客–zookeeper源码分析-事件注册及通信原理</p>
<h2 id="zookeeper基本命令代码"><a href="#zookeeper基本命令代码" class="headerlink" title="zookeeper基本命令代码"></a>zookeeper基本命令代码</h2><h3 id="依赖"><a href="#依赖" class="headerlink" title="依赖"></a>依赖</h3><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line">&lt;groupId&gt;org.apache.zookeeper&lt;/groupId&gt;</span><br><span class="line">&lt;artifactId&gt;zookeeper&lt;/artifactId&gt;</span><br><span class="line">&lt;version&gt;3.5.5&lt;/version&gt;</span><br></pre></td></tr></table></figure>
<h3 id="demo代码"><a href="#demo代码" class="headerlink" title="demo代码"></a>demo代码</h3><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br></pre></td><td class="code"><pre><span class="line">final CountDownLatch coutnDownLatch = new CountDownLatch(1);</span><br><span class="line">ZooKeeper zookeeper = new ZooKeeper(<span class="string">"192.168.156.171:2181,192.168.156.171:2182,192.168.156.171:2183"</span>,</span><br><span class="line">						3000,</span><br><span class="line">						new <span class="function"><span class="title">Watcher</span></span>() &#123;</span><br><span class="line"></span><br><span class="line">							public void process(WatchedEvent event) &#123;</span><br><span class="line">								//注册默认的watcher，当连接上时，会触发该watcher</span><br><span class="line">								//对该zookeeper实例的节点进行操作，watcher填<span class="literal">true</span>使用此默认watcher</span><br><span class="line">								System.out.println(<span class="string">"defaultWatcher process eventType : "</span>+event.getPath()+<span class="string">"-"</span>+event.getType());</span><br><span class="line">								coutnDownLatch.countDown();</span><br><span class="line">							&#125;</span><br><span class="line">	</span><br><span class="line">&#125;);</span><br><span class="line">System.out.println(zookeeper.getState());//connecting---连接中</span><br><span class="line">coutnDownLatch.await();</span><br><span class="line">System.out.println(zookeeper.getState());//connected---已连接</span><br><span class="line">//createMode 模式，PERSISTENT---持久化节点</span><br><span class="line">Stat <span class="built_in">stat</span> = new Stat();</span><br><span class="line">/**</span><br><span class="line"> * 节点状态，包含version</span><br><span class="line"> * create、delete、setData语句会更新节点<span class="built_in">stat</span></span><br><span class="line"> * 并且create、delete子节点会更新父节点的<span class="built_in">stat</span></span><br><span class="line"> * 在create、getData语句传入<span class="built_in">stat</span>，执行后会把最新的<span class="built_in">stat</span>属性复制到传入的<span class="built_in">stat</span>对象中</span><br><span class="line"> * setData、delete语句传入<span class="built_in">stat</span>的version，通过乐观锁验证版本号保证数据同步，版本不对时抛出BadVersionException（如果version是-1，代表不进行版本验证）</span><br><span class="line"> * setData、exists语句返回值为最新的<span class="built_in">stat</span></span><br><span class="line"> */</span><br><span class="line">zookeeper.delete(<span class="string">"/java"</span>,-1);</span><br><span class="line">String path = zookeeper.create(<span class="string">"/java"</span>, <span class="string">"2019-07-29"</span>.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT,<span class="built_in">stat</span>);//返回路径</span><br><span class="line">System.out.println(<span class="string">"path : "</span>+path);</span><br><span class="line">byte[] byteV = zookeeper.getData(path, null, <span class="built_in">stat</span>);</span><br><span class="line">System.out.println(<span class="string">"version:"</span>+stat.getVersion()+<span class="string">"-"</span>+new String(byteV));</span><br><span class="line">/**</span><br><span class="line"> * </span><br><span class="line"> * OPEN_ACL_UNSAFE---完全开放</span><br><span class="line"> * CREATOR_ALL_ACL---创建者拥有权限</span><br><span class="line"> * READ_ACL_UNSAFE---开放读权限</span><br><span class="line"> * ANYONE_ID_UNSAFE---有验证的用户就不被全部权限</span><br><span class="line"> * AUTH_IDS---只对指定用户开放</span><br><span class="line"> */</span><br><span class="line"><span class="built_in">stat</span> = zookeeper.setData(path, <span class="string">"2019-07-30"</span>.getBytes(), stat.getVersion());</span><br><span class="line">zookeeper.getData(path, <span class="literal">true</span>, <span class="built_in">stat</span>);//<span class="literal">true</span>代表注册watcher，使用默认watcher即创建Zookeeper实例时注册的watcher，只响应一次事件</span><br><span class="line">zookeeper.getData(path, new <span class="function"><span class="title">Watcher</span></span>() &#123;//注册专用的watcher，只响应一次事件</span><br><span class="line">	public void process(WatchedEvent event) &#123;</span><br><span class="line">	&#125;</span><br><span class="line">&#125;, <span class="built_in">stat</span>);</span><br><span class="line"></span><br><span class="line">zookeeper.setData(path, <span class="string">"2019-07-30"</span>.getBytes(), stat.getVersion());</span><br><span class="line">byteV = zookeeper.getData(path, null, <span class="built_in">stat</span>);</span><br><span class="line">System.out.println(<span class="string">"version:"</span>+stat.getVersion()+<span class="string">"-"</span>+new String(byteV));</span><br></pre></td></tr></table></figure>
<h2 id="源码分析"><a href="#源码分析" class="headerlink" title="源码分析"></a>源码分析</h2><p>Zookeeper对于集群信息提供jmx的监控支持，详情欢迎查看本人另一篇博客——<a href="https://blog.csdn.net/weixin_44627989/article/details/97636574" target="_blank" rel="noopener">zookeeper源码分析-选举算法</a><br>而对于客户端连接使用的是netty做socket的io处理，实现类为ClientCnxnSocketNetty，还有另一个可选的nio实现类是ClientCnxnSocketNIO<br>Zookeeper构造函数,实例化最终会到下面的构造函数，根据默认配置，创建一个ZKWatchManager对象，将默认wathcer注入该wathcerManager的defaultWatcher成员<br>将该watcherManager以及包含了集群信息的zookeeper对象传递用于实例化一个ClientCnxn对象，并启动<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br></pre></td><td class="code"><pre><span class="line">protected ZKWatchManager <span class="function"><span class="title">defaultWatchManager</span></span>() &#123;</span><br><span class="line">    <span class="built_in">return</span> new ZKWatchManager(getClientConfig().getBoolean(ZKClientConfig.DISABLE_AUTO_WATCH_RESET));</span><br><span class="line">&#125;</span><br><span class="line">public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,</span><br><span class="line">        boolean canBeReadOnly, HostProvider aHostProvider,</span><br><span class="line">        ZKClientConfig clientConfig) throws IOException &#123;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">if</span> (clientConfig == null) &#123;</span><br><span class="line">        clientConfig = new ZKClientConfig();</span><br><span class="line">    &#125;</span><br><span class="line">    this.clientConfig = clientConfig;</span><br><span class="line">    watchManager = defaultWatchManager();</span><br><span class="line">    watchManager.defaultWatcher = watcher;</span><br><span class="line">    ConnectStringParser connectStringParser = new ConnectStringParser(</span><br><span class="line">            connectString);</span><br><span class="line">    hostProvider = aHostProvider;</span><br><span class="line"></span><br><span class="line">    cnxn = createConnection(connectStringParser.getChrootPath(),</span><br><span class="line">            hostProvider, sessionTimeout, this, watchManager,</span><br><span class="line">            getClientCnxnSocket(), canBeReadOnly);</span><br><span class="line">    cnxn.start();</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">// @VisibleForTesting</span><br><span class="line">protected ClientCnxn createConnection(String chrootPath,</span><br><span class="line">        HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,</span><br><span class="line">        ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,</span><br><span class="line">        boolean canBeReadOnly) throws IOException &#123;</span><br><span class="line">    <span class="built_in">return</span> new ClientCnxn(chrootPath, hostProvider, sessionTimeout, this,</span><br><span class="line">            watchManager, clientCnxnSocket, canBeReadOnly);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p>ClientCnxn在实例化时会实例化两个线程成员，作为发送消息的线程以及处理事件的线程<br>因为zookeeper内部使用netty所以发送消息的响应式异步的，当netty收到响应时，将响应封装为一个event，数据的封装返回在evetThread处理<br>要发送的包由一个阻塞队列保存，已发送待确认的包由一个链表保存<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br></pre></td><td class="code"><pre><span class="line">private final LinkedList&lt;Packet&gt; pendingQueue = new LinkedList&lt;Packet&gt;();//等待响应的队列</span><br><span class="line"></span><br><span class="line">private final LinkedBlockingDeque&lt;Packet&gt; outgoingQueue = new LinkedBlockingDeque&lt;Packet&gt;();//要发送的队列</span><br><span class="line"></span><br><span class="line">public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,</span><br><span class="line">        ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,</span><br><span class="line">        long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) &#123;</span><br><span class="line">    this.zooKeeper = zooKeeper;</span><br><span class="line">    this.watcher = watcher;</span><br><span class="line">    this.sessionId = sessionId;</span><br><span class="line">    this.sessionPasswd = sessionPasswd;</span><br><span class="line">    this.sessionTimeout = sessionTimeout;</span><br><span class="line">    this.hostProvider = hostProvider;</span><br><span class="line">    this.chrootPath = chrootPath;</span><br><span class="line"></span><br><span class="line">    connectTimeout = sessionTimeout / hostProvider.size();</span><br><span class="line">    readTimeout = sessionTimeout * 2 / 3;</span><br><span class="line">    readOnly = canBeReadOnly;</span><br><span class="line"></span><br><span class="line">    sendThread = new SendThread(clientCnxnSocket);</span><br><span class="line">    eventThread = new EventThread();</span><br><span class="line">    this.clientConfig=zooKeeper.getClientConfig();</span><br><span class="line">    initRequestTimeout();</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">public void <span class="function"><span class="title">start</span></span>() &#123;</span><br><span class="line">    sendThread.start();</span><br><span class="line">    eventThread.start();</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p>sendThread的处理逻辑为，当服务器状态为活跃时，循环检查计算心跳间隔时间，添加pin包到待发送队列，后面发送以保持连接<br>每一轮最后调用clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this)去开始传输<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br><span class="line">131</span><br><span class="line">132</span><br><span class="line">133</span><br><span class="line">134</span><br><span class="line">135</span><br><span class="line">136</span><br><span class="line">137</span><br><span class="line">138</span><br><span class="line">139</span><br><span class="line">140</span><br><span class="line">141</span><br><span class="line">142</span><br><span class="line">143</span><br><span class="line">144</span><br><span class="line">145</span><br><span class="line">146</span><br><span class="line">147</span><br><span class="line">148</span><br><span class="line">149</span><br><span class="line">150</span><br><span class="line">151</span><br><span class="line">152</span><br><span class="line">153</span><br><span class="line">154</span><br><span class="line">155</span><br><span class="line">156</span><br></pre></td><td class="code"><pre><span class="line">public void <span class="function"><span class="title">run</span></span>() &#123;</span><br><span class="line">           clientCnxnSocket.introduce(this, sessionId, outgoingQueue);</span><br><span class="line">           clientCnxnSocket.updateNow();</span><br><span class="line">           clientCnxnSocket.updateLastSendAndHeard();</span><br><span class="line">           int to;</span><br><span class="line">           long lastPingRwServer = Time.currentElapsedTime();</span><br><span class="line">           final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds</span><br><span class="line">           InetSocketAddress serverAddress = null;</span><br><span class="line">           <span class="keyword">while</span> (state.isAlive()) &#123;</span><br><span class="line">               try &#123;</span><br><span class="line">                   <span class="keyword">if</span> (!clientCnxnSocket.isConnected()) &#123;</span><br><span class="line">                       // don<span class="string">'t re-establish connection if we are closing</span></span><br><span class="line"><span class="string">                       if (closing) &#123;</span></span><br><span class="line"><span class="string">                           break;</span></span><br><span class="line"><span class="string">                       &#125;</span></span><br><span class="line"><span class="string">                       if (rwServerAddress != null) &#123;</span></span><br><span class="line"><span class="string">                           serverAddress = rwServerAddress;</span></span><br><span class="line"><span class="string">                           rwServerAddress = null;</span></span><br><span class="line"><span class="string">                       &#125; else &#123;</span></span><br><span class="line"><span class="string">                           serverAddress = hostProvider.next(1000);</span></span><br><span class="line"><span class="string">                       &#125;</span></span><br><span class="line"><span class="string">                       startConnect(serverAddress);</span></span><br><span class="line"><span class="string">                       clientCnxnSocket.updateLastSendAndHeard();</span></span><br><span class="line"><span class="string">                   &#125;</span></span><br><span class="line"><span class="string"></span></span><br><span class="line"><span class="string">                   if (state.isConnected()) &#123;</span></span><br><span class="line"><span class="string">                       // determine whether we need to send an AuthFailed event.</span></span><br><span class="line"><span class="string">                       if (zooKeeperSaslClient != null) &#123;</span></span><br><span class="line"><span class="string">                           boolean sendAuthEvent = false;</span></span><br><span class="line"><span class="string">                           if (zooKeeperSaslClient.getSaslState() == ZooKeeperSaslClient.SaslState.INITIAL) &#123;</span></span><br><span class="line"><span class="string">                               try &#123;</span></span><br><span class="line"><span class="string">                                   zooKeeperSaslClient.initialize(ClientCnxn.this);</span></span><br><span class="line"><span class="string">                               &#125; catch (SaslException e) &#123;</span></span><br><span class="line"><span class="string">                                  LOG.error("SASL authentication with Zookeeper Quorum member failed: " + e);</span></span><br><span class="line"><span class="string">                                   state = States.AUTH_FAILED;</span></span><br><span class="line"><span class="string">                                   sendAuthEvent = true;</span></span><br><span class="line"><span class="string">                               &#125;</span></span><br><span class="line"><span class="string">                           &#125;</span></span><br><span class="line"><span class="string">                           KeeperState authState = zooKeeperSaslClient.getKeeperState();</span></span><br><span class="line"><span class="string">                           if (authState != null) &#123;</span></span><br><span class="line"><span class="string">                               if (authState == KeeperState.AuthFailed) &#123;</span></span><br><span class="line"><span class="string">                                   // An authentication error occurred during authentication with the Zookeeper Server.</span></span><br><span class="line"><span class="string">                                   state = States.AUTH_FAILED;</span></span><br><span class="line"><span class="string">                                   sendAuthEvent = true;</span></span><br><span class="line"><span class="string">                               &#125; else &#123;</span></span><br><span class="line"><span class="string">                                   if (authState == KeeperState.SaslAuthenticated) &#123;</span></span><br><span class="line"><span class="string">                                       sendAuthEvent = true;</span></span><br><span class="line"><span class="string">                                   &#125;</span></span><br><span class="line"><span class="string">                               &#125;</span></span><br><span class="line"><span class="string">                           &#125;</span></span><br><span class="line"><span class="string"></span></span><br><span class="line"><span class="string">                           if (sendAuthEvent) &#123;</span></span><br><span class="line"><span class="string">                               eventThread.queueEvent(new WatchedEvent(</span></span><br><span class="line"><span class="string">                                     Watcher.Event.EventType.None,</span></span><br><span class="line"><span class="string">                                     authState,null));</span></span><br><span class="line"><span class="string">                               if (state == States.AUTH_FAILED) &#123;</span></span><br><span class="line"><span class="string">                                 eventThread.queueEventOfDeath();</span></span><br><span class="line"><span class="string">                               &#125;</span></span><br><span class="line"><span class="string">                           &#125;</span></span><br><span class="line"><span class="string">                       &#125;</span></span><br><span class="line"><span class="string">                       to = readTimeout - clientCnxnSocket.getIdleRecv();</span></span><br><span class="line"><span class="string">                   &#125; else &#123;</span></span><br><span class="line"><span class="string">                       to = connectTimeout - clientCnxnSocket.getIdleRecv();</span></span><br><span class="line"><span class="string">                   &#125;</span></span><br><span class="line"><span class="string">                   </span></span><br><span class="line"><span class="string">                   if (to &lt;= 0) &#123;</span></span><br><span class="line"><span class="string">                       String warnInfo;</span></span><br><span class="line"><span class="string">                       warnInfo = "Client session timed out, have not heard from server in "</span></span><br><span class="line"><span class="string">                           + clientCnxnSocket.getIdleRecv()</span></span><br><span class="line"><span class="string">                           + "ms"</span></span><br><span class="line"><span class="string">                           + " for sessionid 0x"</span></span><br><span class="line"><span class="string">                           + Long.toHexString(sessionId);</span></span><br><span class="line"><span class="string">                       LOG.warn(warnInfo);</span></span><br><span class="line"><span class="string">                       throw new SessionTimeoutException(warnInfo);</span></span><br><span class="line"><span class="string">                   &#125;</span></span><br><span class="line"><span class="string">                   if (state.isConnected()) &#123;</span></span><br><span class="line"><span class="string">                   	//1000(1 second) is to prevent race condition missing to send the second ping</span></span><br><span class="line"><span class="string">                   	//also make sure not to send too many pings when readTimeout is small </span></span><br><span class="line"><span class="string">                       int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() - </span></span><br><span class="line"><span class="string">                       		((clientCnxnSocket.getIdleSend() &gt; 1000) ? 1000 : 0);</span></span><br><span class="line"><span class="string">                       //send a ping request either time is due or no packet sent out within MAX_SEND_PING_INTERVAL</span></span><br><span class="line"><span class="string">                       if (timeToNextPing &lt;= 0 || clientCnxnSocket.getIdleSend() &gt; MAX_SEND_PING_INTERVAL) &#123;</span></span><br><span class="line"><span class="string">                           sendPing();</span></span><br><span class="line"><span class="string">                           clientCnxnSocket.updateLastSend();</span></span><br><span class="line"><span class="string">                       &#125; else &#123;</span></span><br><span class="line"><span class="string">                           if (timeToNextPing &lt; to) &#123;</span></span><br><span class="line"><span class="string">                               to = timeToNextPing;</span></span><br><span class="line"><span class="string">                           &#125;</span></span><br><span class="line"><span class="string">                       &#125;</span></span><br><span class="line"><span class="string">                   &#125;</span></span><br><span class="line"><span class="string"></span></span><br><span class="line"><span class="string">                   // If we are in read-only mode, seek for read/write server</span></span><br><span class="line"><span class="string">                   if (state == States.CONNECTEDREADONLY) &#123;</span></span><br><span class="line"><span class="string">                       long now = Time.currentElapsedTime();</span></span><br><span class="line"><span class="string">                       int idlePingRwServer = (int) (now - lastPingRwServer);</span></span><br><span class="line"><span class="string">                       if (idlePingRwServer &gt;= pingRwTimeout) &#123;</span></span><br><span class="line"><span class="string">                           lastPingRwServer = now;</span></span><br><span class="line"><span class="string">                           idlePingRwServer = 0;</span></span><br><span class="line"><span class="string">                           pingRwTimeout =</span></span><br><span class="line"><span class="string">                               Math.min(2*pingRwTimeout, maxPingRwTimeout);</span></span><br><span class="line"><span class="string">                           pingRwServer();</span></span><br><span class="line"><span class="string">                       &#125;</span></span><br><span class="line"><span class="string">                       to = Math.min(to, pingRwTimeout - idlePingRwServer);</span></span><br><span class="line"><span class="string">                   &#125;</span></span><br><span class="line"><span class="string"></span></span><br><span class="line"><span class="string">                   clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);</span></span><br><span class="line"><span class="string">               &#125; catch (Throwable e) &#123;</span></span><br><span class="line"><span class="string">                   if (closing) &#123;</span></span><br><span class="line"><span class="string">                       if (LOG.isDebugEnabled()) &#123;</span></span><br><span class="line"><span class="string">                           // closing so this is expected</span></span><br><span class="line"><span class="string">                           LOG.debug("An exception was thrown while closing send thread for session 0x"</span></span><br><span class="line"><span class="string">                                   + Long.toHexString(getSessionId())</span></span><br><span class="line"><span class="string">                                   + " : " + e.getMessage());</span></span><br><span class="line"><span class="string">                       &#125;</span></span><br><span class="line"><span class="string">                       break;</span></span><br><span class="line"><span class="string">                   &#125; else &#123;</span></span><br><span class="line"><span class="string">                       // this is ugly, you have a better way speak up</span></span><br><span class="line"><span class="string">                       if (e instanceof SessionExpiredException) &#123;</span></span><br><span class="line"><span class="string">                           LOG.info(e.getMessage() + ", closing socket connection");</span></span><br><span class="line"><span class="string">                       &#125; else if (e instanceof SessionTimeoutException) &#123;</span></span><br><span class="line"><span class="string">                           LOG.info(e.getMessage() + RETRY_CONN_MSG);</span></span><br><span class="line"><span class="string">                       &#125; else if (e instanceof EndOfStreamException) &#123;</span></span><br><span class="line"><span class="string">                           LOG.info(e.getMessage() + RETRY_CONN_MSG);</span></span><br><span class="line"><span class="string">                       &#125; else if (e instanceof RWServerFoundException) &#123;</span></span><br><span class="line"><span class="string">                           LOG.info(e.getMessage());</span></span><br><span class="line"><span class="string">                       &#125; else if (e instanceof SocketException) &#123;</span></span><br><span class="line"><span class="string">                           LOG.info("Socket error occurred: &#123;&#125;: &#123;&#125;", serverAddress, e.getMessage());</span></span><br><span class="line"><span class="string">                       &#125; else &#123;</span></span><br><span class="line"><span class="string">                           LOG.warn("Session 0x&#123;&#125; for server &#123;&#125;, unexpected error&#123;&#125;",</span></span><br><span class="line"><span class="string">                                           Long.toHexString(getSessionId()),</span></span><br><span class="line"><span class="string">                                           serverAddress,</span></span><br><span class="line"><span class="string">                                           RETRY_CONN_MSG,</span></span><br><span class="line"><span class="string">                                           e);</span></span><br><span class="line"><span class="string">                       &#125;</span></span><br><span class="line"><span class="string">                       // At this point, there might still be new packets appended to outgoingQueue.</span></span><br><span class="line"><span class="string">                       // they will be handled in next connection or cleared up if closed.</span></span><br><span class="line"><span class="string">                       cleanAndNotifyState();</span></span><br><span class="line"><span class="string">                   &#125;</span></span><br><span class="line"><span class="string">               &#125;</span></span><br><span class="line"><span class="string">           &#125;</span></span><br><span class="line"><span class="string">           synchronized (state) &#123;</span></span><br><span class="line"><span class="string">               // When it comes to this point, it guarantees that later queued</span></span><br><span class="line"><span class="string">               // packet to outgoingQueue will be notified of death.</span></span><br><span class="line"><span class="string">               cleanup();</span></span><br><span class="line"><span class="string">           &#125;</span></span><br><span class="line"><span class="string">           clientCnxnSocket.close();</span></span><br><span class="line"><span class="string">           if (state.isAlive()) &#123;</span></span><br><span class="line"><span class="string">               eventThread.queueEvent(new WatchedEvent(Event.EventType.None,</span></span><br><span class="line"><span class="string">                       Event.KeeperState.Disconnected, null));</span></span><br><span class="line"><span class="string">           &#125;</span></span><br><span class="line"><span class="string">           eventThread.queueEvent(new WatchedEvent(Event.EventType.None,</span></span><br><span class="line"><span class="string">                       Event.KeeperState.Closed, null));</span></span><br><span class="line"><span class="string">           ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(),</span></span><br><span class="line"><span class="string">                   "SendThread exited loop for session: 0x"</span></span><br><span class="line"><span class="string">                          + Long.toHexString(getSessionId()));</span></span><br><span class="line"><span class="string">       &#125;</span></span><br></pre></td></tr></table></figure></p>
<p>netty相关的类ClientCnxnSocketNetty中对doTransport的实现为：<br>从outgoingQueue里取出一个待发送的packet，如果不为空就调用doWrite方法去发送数据包，如果当前连接已断开，将包再次加入待发送队列<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br></pre></td><td class="code"><pre><span class="line">void doTransport(int waitTimeOut,</span><br><span class="line">                    List&lt;Packet&gt; pendingQueue,</span><br><span class="line">                    ClientCnxn cnxn)</span><br><span class="line">           throws IOException, InterruptedException &#123;</span><br><span class="line">       try &#123;</span><br><span class="line">           <span class="keyword">if</span> (!firstConnect.await(waitTimeOut, TimeUnit.MILLISECONDS)) &#123;</span><br><span class="line">               <span class="built_in">return</span>;</span><br><span class="line">           &#125;</span><br><span class="line">           Packet head = null;</span><br><span class="line">           <span class="keyword">if</span> (needSasl.get()) &#123;</span><br><span class="line">               <span class="keyword">if</span> (!waitSasl.tryAcquire(waitTimeOut, TimeUnit.MILLISECONDS)) &#123;</span><br><span class="line">                   <span class="built_in">return</span>;</span><br><span class="line">               &#125;</span><br><span class="line">           &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">               head = outgoingQueue.poll(waitTimeOut, TimeUnit.MILLISECONDS);</span><br><span class="line">           &#125;</span><br><span class="line">           // check <span class="keyword">if</span> being waken up on closing.</span><br><span class="line">           <span class="keyword">if</span> (!sendThread.getZkState().isAlive()) &#123;</span><br><span class="line">               // adding back the packet to notify of failure <span class="keyword">in</span> conLossPacket().</span><br><span class="line">               addBack(head);</span><br><span class="line">               <span class="built_in">return</span>;</span><br><span class="line">           &#125;</span><br><span class="line">           // channel disconnection happened</span><br><span class="line">           <span class="keyword">if</span> (disconnected.get()) &#123;</span><br><span class="line">               addBack(head);</span><br><span class="line">               throw new EndOfStreamException(<span class="string">"channel for sessionid 0x"</span></span><br><span class="line">                       + Long.toHexString(sessionId)</span><br><span class="line">                       + <span class="string">" is lost"</span>);</span><br><span class="line">           &#125;</span><br><span class="line">           <span class="keyword">if</span> (head != null) &#123;</span><br><span class="line">               doWrite(pendingQueue, head, cnxn);</span><br><span class="line">           &#125;</span><br><span class="line">       &#125; finally &#123;</span><br><span class="line">           updateNow();</span><br><span class="line">       &#125;</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure></p>
<p>doWrite方法会将除了outgoingQueue里的唤醒包和pin包以及链接登陆的auth包以外的全部包和调用方法带过来的packet加入事件队列<br>（因为pin包和auth包实际没有需要返回的数据信息response回调用方，pin包无处理，auth包只需要失败时的一个响应，收到auth包的失败信息，添加一个authFailed的事件到事件队列，源码里也是这么做的*在sendThread的readResponse方法）<br>并且outgoingQueue里的packet，除了唤醒包以外，pin包、auth包以及剩下的packet和调用方法带过来的packet都会被发送<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br></pre></td><td class="code"><pre><span class="line">@Override</span><br><span class="line">	private void doWrite(List&lt;Packet&gt; pendingQueue, Packet p, ClientCnxn cnxn) &#123;</span><br><span class="line">        updateNow();</span><br><span class="line">        boolean anyPacketsSent = <span class="literal">false</span>;</span><br><span class="line">        <span class="keyword">while</span> (<span class="literal">true</span>) &#123;</span><br><span class="line">            <span class="keyword">if</span> (p != WakeupPacket.getInstance()) &#123;</span><br><span class="line">                <span class="keyword">if</span> ((p.requestHeader != null) &amp;&amp;</span><br><span class="line">                        (p.requestHeader.getType() != ZooDefs.OpCode.ping) &amp;&amp;</span><br><span class="line">                        (p.requestHeader.getType() != ZooDefs.OpCode.auth)) &#123;</span><br><span class="line">                    p.requestHeader.setXid(cnxn.getXid());</span><br><span class="line">                    synchronized (pendingQueue) &#123;</span><br><span class="line">                        pendingQueue.add(p);</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">                sendPktOnly(p);</span><br><span class="line">                anyPacketsSent = <span class="literal">true</span>;</span><br><span class="line">            &#125;</span><br><span class="line">            <span class="keyword">if</span> (outgoingQueue.isEmpty()) &#123;</span><br><span class="line">                <span class="built_in">break</span>;</span><br><span class="line">            &#125;</span><br><span class="line">            p = outgoingQueue.remove();</span><br><span class="line">        &#125;</span><br><span class="line">        // TODO: maybe we should flush <span class="keyword">in</span> the loop above every N packets/bytes?</span><br><span class="line">        // But, how <span class="keyword">do</span> we determine the right value <span class="keyword">for</span> N ...</span><br><span class="line">        <span class="keyword">if</span> (anyPacketsSent) &#123;</span><br><span class="line">            channel.flush();</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure></p>
<p>当netty的channel读到数据时，会调用sendThread的readResponse方法装载响应，最后会调用wakeupCnxn往outgoingQueue里添加唤醒包<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br></pre></td><td class="code"><pre><span class="line">protected void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) throws Exception &#123;</span><br><span class="line">	updateNow();</span><br><span class="line">	<span class="keyword">while</span> (buf.isReadable()) &#123;</span><br><span class="line">		<span class="keyword">if</span> (incomingBuffer.remaining() &gt; buf.readableBytes()) &#123;</span><br><span class="line">			int newLimit = incomingBuffer.position()</span><br><span class="line">					+ buf.readableBytes();</span><br><span class="line">			incomingBuffer.limit(newLimit);</span><br><span class="line">		&#125;</span><br><span class="line">		buf.readBytes(incomingBuffer);</span><br><span class="line">		incomingBuffer.limit(incomingBuffer.capacity());</span><br><span class="line"></span><br><span class="line">		<span class="keyword">if</span> (!incomingBuffer.hasRemaining()) &#123;</span><br><span class="line">			incomingBuffer.flip();</span><br><span class="line">			<span class="keyword">if</span> (incomingBuffer == lenBuffer) &#123;</span><br><span class="line">				recvCount.getAndIncrement();</span><br><span class="line">				readLength();</span><br><span class="line">			&#125; <span class="keyword">else</span> <span class="keyword">if</span> (!initialized) &#123;</span><br><span class="line">				readConnectResult();</span><br><span class="line">				lenBuffer.clear();</span><br><span class="line">				incomingBuffer = lenBuffer;</span><br><span class="line">				initialized = <span class="literal">true</span>;</span><br><span class="line">				updateLastHeard();</span><br><span class="line">			&#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				sendThread.readResponse(incomingBuffer);</span><br><span class="line">				lenBuffer.clear();</span><br><span class="line">				incomingBuffer = lenBuffer;</span><br><span class="line">				updateLastHeard();</span><br><span class="line">			&#125;</span><br><span class="line">		&#125;</span><br><span class="line">	&#125;</span><br><span class="line">	wakeupCnxn();</span><br><span class="line">	// Note: SimpleChannelInboundHandler releases the ByteBuf <span class="keyword">for</span> us</span><br><span class="line">	// so we don<span class="string">'t need to do it.</span></span><br><span class="line"><span class="string">   &#125;</span></span><br><span class="line"><span class="string">   private void wakeupCnxn() &#123;</span></span><br><span class="line"><span class="string">       if (needSasl.get()) &#123;</span></span><br><span class="line"><span class="string">           waitSasl.release();</span></span><br><span class="line"><span class="string">       &#125;</span></span><br><span class="line"><span class="string">       outgoingQueue.add(WakeupPacket.getInstance());</span></span><br><span class="line"><span class="string">   &#125;</span></span><br></pre></td></tr></table></figure></p>
<p>sendThread的readResponse方法根据响应的数据报类型做处理<br>如果是pin包不做处理只做日志，并返回<br>如果是auth包，做日志处理，判断err码，如果auth失败了，就往事件队列里增加一个type为None，stat为AuthFailed的事件，并返回<br>如果是主动监听的服务端通知事件，创建一个WatcherEvent反序列化response的信息，并用一个WatchedEvent包装以后，加入事件队列，并返回<br>如果不满足以上的，先判断如果当前正在进行SASL身份验证，构造和立即发送响应数据包，而不是排队响应其他数据包<br>剩下的进行普通数据报处理，从pendingQueue的首部取出一个已发送待响应的packet（因为是从头拿，所以是最早发送的带响应packet）<br>判断取出的packet与读到的ReplyHeader数据的xid是否相等，不相等说明出现丢包，抛出异常<br>xid相等则将本次数据的response信息反序列化装载到packet的resonse成员<br>无论是否丢包，最终会调用finishPacket(packet)对packet做善后工作<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br></pre></td><td class="code"><pre><span class="line">void readResponse(ByteBuffer incomingBuffer) throws IOException &#123;</span><br><span class="line">           ByteBufferInputStream bbis = new ByteBufferInputStream(</span><br><span class="line">                   incomingBuffer);</span><br><span class="line">           BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);</span><br><span class="line">           ReplyHeader replyHdr = new ReplyHeader();</span><br><span class="line"></span><br><span class="line">           replyHdr.deserialize(bbia, <span class="string">"header"</span>);</span><br><span class="line">           <span class="keyword">if</span> (replyHdr.getXid() == -2) &#123;</span><br><span class="line">               // -2 is the xid <span class="keyword">for</span> pings</span><br><span class="line">               <span class="keyword">if</span> (LOG.isDebugEnabled()) &#123;</span><br><span class="line">                   LOG.debug(<span class="string">"Got ping response for sessionid: 0x"</span></span><br><span class="line">                           + Long.toHexString(sessionId)</span><br><span class="line">                           + <span class="string">" after "</span></span><br><span class="line">                           + ((System.nanoTime() - lastPingSentNs) / 1000000)</span><br><span class="line">                           + <span class="string">"ms"</span>);</span><br><span class="line">               &#125;</span><br><span class="line">               <span class="built_in">return</span>;</span><br><span class="line">           &#125;</span><br><span class="line">           <span class="keyword">if</span> (replyHdr.getXid() == -4) &#123;</span><br><span class="line">               // -4 is the xid <span class="keyword">for</span> AuthPacket               </span><br><span class="line">               <span class="keyword">if</span>(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) &#123;</span><br><span class="line">                   state = States.AUTH_FAILED;                    </span><br><span class="line">                   eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None, </span><br><span class="line">                           Watcher.Event.KeeperState.AuthFailed, null) );</span><br><span class="line">                   eventThread.queueEventOfDeath();</span><br><span class="line">               &#125;</span><br><span class="line">               <span class="keyword">if</span> (LOG.isDebugEnabled()) &#123;</span><br><span class="line">                   LOG.debug(<span class="string">"Got auth sessionid:0x"</span></span><br><span class="line">                           + Long.toHexString(sessionId));</span><br><span class="line">               &#125;</span><br><span class="line">               <span class="built_in">return</span>;</span><br><span class="line">           &#125;</span><br><span class="line">           <span class="keyword">if</span> (replyHdr.getXid() == -1) &#123;</span><br><span class="line">               // -1 means notification</span><br><span class="line">               <span class="keyword">if</span> (LOG.isDebugEnabled()) &#123;</span><br><span class="line">                   LOG.debug(<span class="string">"Got notification sessionid:0x"</span></span><br><span class="line">                       + Long.toHexString(sessionId));</span><br><span class="line">               &#125;</span><br><span class="line">               WatcherEvent event = new WatcherEvent();</span><br><span class="line">               event.deserialize(bbia, <span class="string">"response"</span>);</span><br><span class="line"></span><br><span class="line">               // convert from a server path to a client path</span><br><span class="line">               <span class="keyword">if</span> (chrootPath != null) &#123;</span><br><span class="line">                   String serverPath = event.getPath();</span><br><span class="line">                   <span class="keyword">if</span>(serverPath.compareTo(chrootPath)==0)</span><br><span class="line">                       event.setPath(<span class="string">"/"</span>);</span><br><span class="line">                   <span class="keyword">else</span> <span class="keyword">if</span> (serverPath.length() &gt; chrootPath.length())</span><br><span class="line">                       event.setPath(serverPath.substring(chrootPath.length()));</span><br><span class="line">                   <span class="keyword">else</span> &#123;</span><br><span class="line">                   	LOG.warn(<span class="string">"Got server path "</span> + event.getPath()</span><br><span class="line">                   			+ <span class="string">" which is too short for chroot path "</span></span><br><span class="line">                   			+ chrootPath);</span><br><span class="line">                   &#125;</span><br><span class="line">               &#125;</span><br><span class="line"></span><br><span class="line">               WatchedEvent we = new WatchedEvent(event);</span><br><span class="line">               <span class="keyword">if</span> (LOG.isDebugEnabled()) &#123;</span><br><span class="line">                   LOG.debug(<span class="string">"Got "</span> + we + <span class="string">" for sessionid 0x"</span></span><br><span class="line">                           + Long.toHexString(sessionId));</span><br><span class="line">               &#125;</span><br><span class="line"></span><br><span class="line">               eventThread.queueEvent( we );</span><br><span class="line">               <span class="built_in">return</span>;</span><br><span class="line">           &#125;</span><br><span class="line"></span><br><span class="line">           // If SASL authentication is currently <span class="keyword">in</span> progress, construct and</span><br><span class="line">           // send a response packet immediately, rather than queuing a</span><br><span class="line">           // response as with other packets.</span><br><span class="line">           <span class="keyword">if</span> (tunnelAuthInProgress()) &#123;</span><br><span class="line">               GetSASLRequest request = new GetSASLRequest();</span><br><span class="line">               request.deserialize(bbia,<span class="string">"token"</span>);</span><br><span class="line">               zooKeeperSaslClient.respondToServer(request.getToken(),</span><br><span class="line">                 ClientCnxn.this);</span><br><span class="line">               <span class="built_in">return</span>;</span><br><span class="line">           &#125;</span><br><span class="line"></span><br><span class="line">           Packet packet;</span><br><span class="line">           synchronized (pendingQueue) &#123;</span><br><span class="line">               <span class="keyword">if</span> (pendingQueue.size() == 0) &#123;</span><br><span class="line">                   throw new IOException(<span class="string">"Nothing in the queue, but got "</span></span><br><span class="line">                           + replyHdr.getXid());</span><br><span class="line">               &#125;</span><br><span class="line">               packet = pendingQueue.remove();</span><br><span class="line">           &#125;</span><br><span class="line">           /*</span><br><span class="line">            * Since requests are processed <span class="keyword">in</span> order, we better get a response</span><br><span class="line">            * to the first request!</span><br><span class="line">            */</span><br><span class="line">           try &#123;</span><br><span class="line">               <span class="keyword">if</span> (packet.requestHeader.getXid() != replyHdr.getXid()) &#123;</span><br><span class="line">                   packet.replyHeader.setErr(</span><br><span class="line">                           KeeperException.Code.CONNECTIONLOSS.intValue());</span><br><span class="line">                   throw new IOException(<span class="string">"Xid out of order. Got Xid "</span></span><br><span class="line">                           + replyHdr.getXid() + <span class="string">" with err "</span> +</span><br><span class="line">                           + replyHdr.getErr() +</span><br><span class="line">                           <span class="string">" expected Xid "</span></span><br><span class="line">                           + packet.requestHeader.getXid()</span><br><span class="line">                           + <span class="string">" for a packet with details: "</span></span><br><span class="line">                           + packet );</span><br><span class="line">               &#125;</span><br><span class="line"></span><br><span class="line">               packet.replyHeader.setXid(replyHdr.getXid());</span><br><span class="line">               packet.replyHeader.setErr(replyHdr.getErr());</span><br><span class="line">               packet.replyHeader.setZxid(replyHdr.getZxid());</span><br><span class="line">               <span class="keyword">if</span> (replyHdr.getZxid() &gt; 0) &#123;</span><br><span class="line">                   lastZxid = replyHdr.getZxid();</span><br><span class="line">               &#125;</span><br><span class="line">               <span class="keyword">if</span> (packet.response != null &amp;&amp; replyHdr.getErr() == 0) &#123;</span><br><span class="line">                   packet.response.deserialize(bbia, <span class="string">"response"</span>);</span><br><span class="line">               &#125;</span><br><span class="line"></span><br><span class="line">               <span class="keyword">if</span> (LOG.isDebugEnabled()) &#123;</span><br><span class="line">                   LOG.debug(<span class="string">"Reading reply sessionid:0x"</span></span><br><span class="line">                           + Long.toHexString(sessionId) + <span class="string">", packet:: "</span> + packet);</span><br><span class="line">               &#125;</span><br><span class="line">           &#125; finally &#123;</span><br><span class="line">               finishPacket(packet);</span><br><span class="line">           &#125;</span><br><span class="line">       &#125;</span><br></pre></td></tr></table></figure></p>
<p>在eventThread的queueEvent方法里，会调用watcher.materialize(event.getState(),event.getType(), event.getPath())识别事件<br>构建一个WatcherSetEventPair加入事件队列waitingEvents<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br></pre></td><td class="code"><pre><span class="line">private final LinkedBlockingQueue&lt;Object&gt; waitingEvents =</span><br><span class="line">           new LinkedBlockingQueue&lt;Object&gt;();</span><br><span class="line">public void queueEvent(WatchedEvent event) &#123;</span><br><span class="line">	queueEvent(event, null);</span><br><span class="line">&#125;</span><br><span class="line">private void queueEvent(WatchedEvent event,</span><br><span class="line">		Set&lt;Watcher&gt; materializedWatchers) &#123;</span><br><span class="line">	<span class="keyword">if</span> (event.getType() == EventType.None</span><br><span class="line">			&amp;&amp; sessionState == event.getState()) &#123;</span><br><span class="line">		<span class="built_in">return</span>;</span><br><span class="line">	&#125;</span><br><span class="line">	sessionState = event.getState();</span><br><span class="line">	final Set&lt;Watcher&gt; watchers;</span><br><span class="line">	<span class="keyword">if</span> (materializedWatchers == null) &#123;</span><br><span class="line">		// materialize the watchers based on the event</span><br><span class="line">		watchers = watcher.materialize(event.getState(),</span><br><span class="line">				event.getType(), event.getPath());</span><br><span class="line">	&#125; <span class="keyword">else</span> &#123;</span><br><span class="line">		watchers = new HashSet&lt;Watcher&gt;();</span><br><span class="line">		watchers.addAll(materializedWatchers);</span><br><span class="line">	&#125;</span><br><span class="line">	WatcherSetEventPair pair = new WatcherSetEventPair(watchers, event);</span><br><span class="line">	// queue the pair (watch <span class="built_in">set</span> &amp; event) <span class="keyword">for</span> later processing</span><br><span class="line">	waitingEvents.add(pair);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p>eventThread使用的watcher实际为Zookeeper类传递给ClientCnxn的watcherManager成员<br>也就是zookeeper本地注册watcher的对象<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line">protected ClientCnxn createConnection(String chrootPath,</span><br><span class="line">        HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,</span><br><span class="line">        ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,</span><br><span class="line">        boolean canBeReadOnly) throws IOException &#123;</span><br><span class="line">    <span class="built_in">return</span> new ClientCnxn(chrootPath, hostProvider, sessionTimeout, this,</span><br><span class="line">            watchManager, clientCnxnSocket, canBeReadOnly);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br></pre></td><td class="code"><pre><span class="line">   public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,</span><br><span class="line">           ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket, boolean canBeReadOnly)</span><br><span class="line">           throws IOException &#123;</span><br><span class="line">       this(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher,</span><br><span class="line">            clientCnxnSocket, 0, new byte[16], canBeReadOnly);</span><br><span class="line">   &#125;</span><br><span class="line">public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,</span><br><span class="line">           ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,</span><br><span class="line">           long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) &#123;</span><br><span class="line">       this.zooKeeper = zooKeeper;</span><br><span class="line">       this.watcher = watcher;</span><br><span class="line">       this.sessionId = sessionId;</span><br><span class="line">       this.sessionPasswd = sessionPasswd;</span><br><span class="line">       this.sessionTimeout = sessionTimeout;</span><br><span class="line">       this.hostProvider = hostProvider;</span><br><span class="line">       this.chrootPath = chrootPath;</span><br><span class="line"></span><br><span class="line">       connectTimeout = sessionTimeout / hostProvider.size();</span><br><span class="line">       readTimeout = sessionTimeout * 2 / 3;</span><br><span class="line">       readOnly = canBeReadOnly;</span><br><span class="line"></span><br><span class="line">       sendThread = new SendThread(clientCnxnSocket);</span><br><span class="line">       eventThread = new EventThread();</span><br><span class="line">       this.clientConfig=zooKeeper.getClientConfig();</span><br><span class="line">       initRequestTimeout();</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure>
<p>ZkWatcherManager使用三个集合成员存储不同事件类型的watcher，并设置一个默认的watcher<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line">static class ZKWatchManager implements ClientWatchManager &#123;</span><br><span class="line">       private final Map&lt;String, Set&lt;Watcher&gt;&gt; dataWatches =</span><br><span class="line">           new HashMap&lt;String, Set&lt;Watcher&gt;&gt;();</span><br><span class="line">       private final Map&lt;String, Set&lt;Watcher&gt;&gt; existWatches =</span><br><span class="line">           new HashMap&lt;String, Set&lt;Watcher&gt;&gt;();</span><br><span class="line">       private final Map&lt;String, Set&lt;Watcher&gt;&gt; childWatches =</span><br><span class="line">           new HashMap&lt;String, Set&lt;Watcher&gt;&gt;();</span><br><span class="line">       private boolean disableAutoWatchReset;</span><br><span class="line"></span><br><span class="line">       ZKWatchManager(boolean disableAutoWatchReset) &#123;</span><br><span class="line">           this.disableAutoWatchReset = disableAutoWatchReset;</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       protected volatile Watcher defaultWatcher;</span><br></pre></td></tr></table></figure></p>
<p>其materialize方法根据事件类型和路径信息，匹配到对应的watcher事件并返回事件集合，可以看到使用remove方法获得事件集合。这也是为什么注册的事件只响应一次<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br></pre></td><td class="code"><pre><span class="line">public Set&lt;Watcher&gt; materialize(Watcher.Event.KeeperState state,</span><br><span class="line">                                       Watcher.Event.EventType <span class="built_in">type</span>,</span><br><span class="line">                                       String clientPath)</span><br><span class="line">       &#123;</span><br><span class="line">           Set&lt;Watcher&gt; result = new HashSet&lt;Watcher&gt;();</span><br><span class="line"></span><br><span class="line">           switch (<span class="built_in">type</span>) &#123;</span><br><span class="line">           <span class="keyword">case</span> None:</span><br><span class="line">               result.add(defaultWatcher);</span><br><span class="line">               boolean clear = disableAutoWatchReset &amp;&amp; state != Watcher.Event.KeeperState.SyncConnected;</span><br><span class="line">               synchronized(dataWatches) &#123;</span><br><span class="line">                   <span class="keyword">for</span>(Set&lt;Watcher&gt; ws: dataWatches.values()) &#123;</span><br><span class="line">                       result.addAll(ws);</span><br><span class="line">                   &#125;</span><br><span class="line">                   <span class="keyword">if</span> (clear) &#123;</span><br><span class="line">                       dataWatches.clear();</span><br><span class="line">                   &#125;</span><br><span class="line">               &#125;</span><br><span class="line"></span><br><span class="line">               synchronized(existWatches) &#123;</span><br><span class="line">                   <span class="keyword">for</span>(Set&lt;Watcher&gt; ws: existWatches.values()) &#123;</span><br><span class="line">                       result.addAll(ws);</span><br><span class="line">                   &#125;</span><br><span class="line">                   <span class="keyword">if</span> (clear) &#123;</span><br><span class="line">                       existWatches.clear();</span><br><span class="line">                   &#125;</span><br><span class="line">               &#125;</span><br><span class="line"></span><br><span class="line">               synchronized(childWatches) &#123;</span><br><span class="line">                   <span class="keyword">for</span>(Set&lt;Watcher&gt; ws: childWatches.values()) &#123;</span><br><span class="line">                       result.addAll(ws);</span><br><span class="line">                   &#125;</span><br><span class="line">                   <span class="keyword">if</span> (clear) &#123;</span><br><span class="line">                       childWatches.clear();</span><br><span class="line">                   &#125;</span><br><span class="line">               &#125;</span><br><span class="line"></span><br><span class="line">               <span class="built_in">return</span> result;</span><br><span class="line">           <span class="keyword">case</span> NodeDataChanged:</span><br><span class="line">           <span class="keyword">case</span> NodeCreated:</span><br><span class="line">               synchronized (dataWatches) &#123;</span><br><span class="line">                   addTo(dataWatches.remove(clientPath), result);</span><br><span class="line">               &#125;</span><br><span class="line">               synchronized (existWatches) &#123;</span><br><span class="line">                   addTo(existWatches.remove(clientPath), result);</span><br><span class="line">               &#125;</span><br><span class="line">               <span class="built_in">break</span>;</span><br><span class="line">           <span class="keyword">case</span> NodeChildrenChanged:</span><br><span class="line">               synchronized (childWatches) &#123;</span><br><span class="line">                   addTo(childWatches.remove(clientPath), result);</span><br><span class="line">               &#125;</span><br><span class="line">               <span class="built_in">break</span>;</span><br><span class="line">           <span class="keyword">case</span> NodeDeleted:</span><br><span class="line">               synchronized (dataWatches) &#123;</span><br><span class="line">                   addTo(dataWatches.remove(clientPath), result);</span><br><span class="line">               &#125;</span><br><span class="line">               // XXX This shouldn<span class="string">'t be needed, but just in case</span></span><br><span class="line"><span class="string">               synchronized (existWatches) &#123;</span></span><br><span class="line"><span class="string">                   Set&lt;Watcher&gt; list = existWatches.remove(clientPath);</span></span><br><span class="line"><span class="string">                   if (list != null) &#123;</span></span><br><span class="line"><span class="string">                       addTo(list, result);</span></span><br><span class="line"><span class="string">                       LOG.warn("We are triggering an exists watch for delete! Shouldn'</span>t happen!<span class="string">");</span></span><br><span class="line"><span class="string">                   &#125;</span></span><br><span class="line"><span class="string">               &#125;</span></span><br><span class="line"><span class="string">               synchronized (childWatches) &#123;</span></span><br><span class="line"><span class="string">                   addTo(childWatches.remove(clientPath), result);</span></span><br><span class="line"><span class="string">               &#125;</span></span><br><span class="line"><span class="string">               break;</span></span><br><span class="line"><span class="string">           default:</span></span><br><span class="line"><span class="string">               String msg = "</span>Unhandled watch event <span class="built_in">type</span> <span class="string">" + type</span></span><br><span class="line"><span class="string">                   + "</span> with state <span class="string">" + state + "</span> on path <span class="string">" + clientPath;</span></span><br><span class="line"><span class="string">               LOG.error(msg);</span></span><br><span class="line"><span class="string">               throw new RuntimeException(msg);</span></span><br><span class="line"><span class="string">           &#125;</span></span><br><span class="line"><span class="string"></span></span><br><span class="line"><span class="string">           return result;</span></span><br><span class="line"><span class="string">       &#125;</span></span><br></pre></td></tr></table></figure></p>
<p>说回readResponse最后对数据报的处理，在完成将数据报对应的本地注册事件加入事件队列以后，执行finishPacket方法<br>在finishPacket中，设置packet的finished标志位true<br>如果该packet没有绑定异步回调函数，则直接调用packet的notify唤醒（因为是异步发送，前面获得packet以后会陷入阻塞wait）<br>否则调用eventThread.queuePacket方法对响应后的数据包做进一步处理<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br></pre></td><td class="code"><pre><span class="line">protected void finishPacket(Packet p) &#123;</span><br><span class="line">       int err = p.replyHeader.getErr();</span><br><span class="line">       <span class="keyword">if</span> (p.watchRegistration != null) &#123;</span><br><span class="line">           p.watchRegistration.register(err);</span><br><span class="line">       &#125;</span><br><span class="line">       // Add all the removed watch events to the event queue, so that the</span><br><span class="line">       // clients will be notified with <span class="string">'Data/Child WatchRemoved'</span> event <span class="built_in">type</span>.</span><br><span class="line">       <span class="keyword">if</span> (p.watchDeregistration != null) &#123;</span><br><span class="line">           Map&lt;EventType, Set&lt;Watcher&gt;&gt; materializedWatchers = null;</span><br><span class="line">           try &#123;</span><br><span class="line">               materializedWatchers = p.watchDeregistration.unregister(err);</span><br><span class="line">               <span class="keyword">for</span> (Entry&lt;EventType, Set&lt;Watcher&gt;&gt; entry : materializedWatchers</span><br><span class="line">                       .entrySet()) &#123;</span><br><span class="line">                   Set&lt;Watcher&gt; watchers = entry.getValue();</span><br><span class="line">                   <span class="keyword">if</span> (watchers.size() &gt; 0) &#123;</span><br><span class="line">                       queueEvent(p.watchDeregistration.getClientPath(), err,</span><br><span class="line">                               watchers, entry.getKey());</span><br><span class="line">                       // ignore connectionloss when removing from <span class="built_in">local</span></span><br><span class="line">                       // session</span><br><span class="line">                       p.replyHeader.setErr(Code.OK.intValue());</span><br><span class="line">                   &#125;</span><br><span class="line">               &#125;</span><br><span class="line">           &#125; catch (KeeperException.NoWatcherException nwe) &#123;</span><br><span class="line">               p.replyHeader.setErr(nwe.code().intValue());</span><br><span class="line">           &#125; catch (KeeperException ke) &#123;</span><br><span class="line">               p.replyHeader.setErr(ke.code().intValue());</span><br><span class="line">           &#125;</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       <span class="keyword">if</span> (p.cb == null) &#123;</span><br><span class="line">           synchronized (p) &#123;</span><br><span class="line">               p.finished = <span class="literal">true</span>;</span><br><span class="line">               p.notifyAll();</span><br><span class="line">           &#125;</span><br><span class="line">       &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">           p.finished = <span class="literal">true</span>;</span><br><span class="line">           eventThread.queuePacket(p);</span><br><span class="line">       &#125;</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure></p>
<p>eventThread的queuePacket方法先判断当前是否已经被要求结束，如果是，判断是否还没有停止运行，没有则将packet加入待处理队列，否则直接调用processEvent处理packet<br>（因为在设置wasKilled为true后，run方法里会将waitingEvents里事件处理完后才设置停止运行并结束线程）<br>如果wasKilled不为true，处于运行中，直接将packet加入waitingEvents<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"></span><br><span class="line">public void queuePacket(Packet packet) &#123;</span><br><span class="line">  <span class="keyword">if</span> (wasKilled) &#123;</span><br><span class="line">	 synchronized (waitingEvents) &#123;</span><br><span class="line">		<span class="keyword">if</span> (isRunning) waitingEvents.add(packet);</span><br><span class="line">		<span class="keyword">else</span> processEvent(packet);</span><br><span class="line">	 &#125;</span><br><span class="line">  &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">	 waitingEvents.add(packet);</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p>eventThread的run方法，循环从waitingEvents里（事件队列，是个阻塞队列）取出待处理的事件<br>调用processEvent去处理事件，如果收到结束事件（eventOfDeath），设置wasKilled为true，在处理完waitingEvents的事件后，停止运行（isRunnig设置为false）<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br></pre></td><td class="code"><pre><span class="line">public void <span class="function"><span class="title">run</span></span>() &#123;</span><br><span class="line">   try &#123;</span><br><span class="line">	  isRunning = <span class="literal">true</span>;</span><br><span class="line">	  <span class="keyword">while</span> (<span class="literal">true</span>) &#123;</span><br><span class="line">		 Object event = waitingEvents.take();</span><br><span class="line">		 <span class="keyword">if</span> (event == eventOfDeath) &#123;</span><br><span class="line">			wasKilled = <span class="literal">true</span>;</span><br><span class="line">		 &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">			processEvent(event);</span><br><span class="line">		 &#125;</span><br><span class="line">		 <span class="keyword">if</span> (wasKilled)</span><br><span class="line">			synchronized (waitingEvents) &#123;</span><br><span class="line">			   <span class="keyword">if</span> (waitingEvents.isEmpty()) &#123;</span><br><span class="line">				  isRunning = <span class="literal">false</span>;</span><br><span class="line">				  <span class="built_in">break</span>;</span><br><span class="line">			   &#125;</span><br><span class="line">			&#125;</span><br><span class="line">	  &#125;</span><br><span class="line">   &#125; catch (InterruptedException e) &#123;</span><br><span class="line">	  LOG.error(<span class="string">"Event thread exiting due to interruption"</span>, e);</span><br><span class="line">   &#125;</span><br><span class="line"></span><br><span class="line">	LOG.info(<span class="string">"EventThread shut down for session: 0x&#123;&#125;"</span>,</span><br><span class="line">			 Long.toHexString(getSessionId()));</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p>在查看processEvent方法前，我们先回过头来看执行各个语句的源码</p>
<h3 id="create"><a href="#create" class="headerlink" title="create"></a>create</h3><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br></pre></td><td class="code"><pre><span class="line">public String create(final String path, byte data[], List&lt;ACL&gt; acl,</span><br><span class="line">           CreateMode createMode, Stat <span class="built_in">stat</span>, long ttl)</span><br><span class="line">           throws KeeperException, InterruptedException &#123;</span><br><span class="line">       final String clientPath = path;</span><br><span class="line">       PathUtils.validatePath(clientPath, createMode.isSequential());</span><br><span class="line">       EphemeralType.validateTTL(createMode, ttl);</span><br><span class="line"></span><br><span class="line">       final String serverPath = prependChroot(clientPath);</span><br><span class="line"></span><br><span class="line">       RequestHeader h = new RequestHeader();</span><br><span class="line">       setCreateHeader(createMode, h);</span><br><span class="line">       Create2Response response = new Create2Response();</span><br><span class="line">       <span class="keyword">if</span> (acl != null &amp;&amp; acl.size() == 0) &#123;</span><br><span class="line">           throw new KeeperException.InvalidACLException();</span><br><span class="line">       &#125;</span><br><span class="line">       Record record = makeCreateRecord(createMode, serverPath, data, acl, ttl);</span><br><span class="line">       ReplyHeader r = cnxn.submitRequest(h, record, response, null);</span><br><span class="line">       <span class="keyword">if</span> (r.getErr() != 0) &#123;</span><br><span class="line">           throw KeeperException.create(KeeperException.Code.get(r.getErr()),</span><br><span class="line">                   clientPath);</span><br><span class="line">       &#125;</span><br><span class="line">       <span class="keyword">if</span> (<span class="built_in">stat</span> != null) &#123;</span><br><span class="line">           DataTree.copyStat(response.getStat(), <span class="built_in">stat</span>);</span><br><span class="line">       &#125;</span><br><span class="line">       <span class="keyword">if</span> (cnxn.chrootPath == null) &#123;</span><br><span class="line">           <span class="built_in">return</span> response.getPath();</span><br><span class="line">       &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">           <span class="built_in">return</span> response.getPath().substring(cnxn.chrootPath.length());</span><br><span class="line">       &#125;</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure>
<h3 id="setData-带回调"><a href="#setData-带回调" class="headerlink" title="setData(带回调)"></a>setData(带回调)</h3><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line">public void setData(final String path, byte data[], int version,</span><br><span class="line">           StatCallback cb, Object ctx)</span><br><span class="line">   &#123;</span><br><span class="line">       final String clientPath = path;</span><br><span class="line">       PathUtils.validatePath(clientPath);</span><br><span class="line"></span><br><span class="line">       final String serverPath = prependChroot(clientPath);</span><br><span class="line"></span><br><span class="line">       RequestHeader h = new RequestHeader();</span><br><span class="line">       h.setType(ZooDefs.OpCode.setData);</span><br><span class="line">       SetDataRequest request = new SetDataRequest();</span><br><span class="line">       request.setPath(serverPath);</span><br><span class="line">       request.setData(data);</span><br><span class="line">       request.setVersion(version);</span><br><span class="line">       SetDataResponse response = new SetDataResponse();</span><br><span class="line">       cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,</span><br><span class="line">               clientPath, serverPath, ctx, null);</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure>
<h3 id="getData"><a href="#getData" class="headerlink" title="getData"></a>getData</h3><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br></pre></td><td class="code"><pre><span class="line">public byte[] getData(final String path, Watcher watcher, Stat <span class="built_in">stat</span>)</span><br><span class="line">       throws KeeperException, InterruptedException</span><br><span class="line">    &#123;</span><br><span class="line">       final String clientPath = path;</span><br><span class="line">       PathUtils.validatePath(clientPath);</span><br><span class="line"></span><br><span class="line">       // the watch contains the un-chroot path</span><br><span class="line">       WatchRegistration wcb = null;</span><br><span class="line">       <span class="keyword">if</span> (watcher != null) &#123;</span><br><span class="line">           wcb = new DataWatchRegistration(watcher, clientPath);</span><br><span class="line">       &#125;</span><br><span class="line"></span><br><span class="line">       final String serverPath = prependChroot(clientPath);</span><br><span class="line"></span><br><span class="line">       RequestHeader h = new RequestHeader();</span><br><span class="line">       h.setType(ZooDefs.OpCode.getData);</span><br><span class="line">       GetDataRequest request = new GetDataRequest();</span><br><span class="line">       request.setPath(serverPath);</span><br><span class="line">       request.setWatch(watcher != null);</span><br><span class="line">       GetDataResponse response = new GetDataResponse();</span><br><span class="line">       ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);</span><br><span class="line">       <span class="keyword">if</span> (r.getErr() != 0) &#123;</span><br><span class="line">           throw KeeperException.create(KeeperException.Code.get(r.getErr()),</span><br><span class="line">                   clientPath);</span><br><span class="line">       &#125;</span><br><span class="line">       <span class="keyword">if</span> (<span class="built_in">stat</span> != null) &#123;</span><br><span class="line">           DataTree.copyStat(response.getStat(), <span class="built_in">stat</span>);</span><br><span class="line">       &#125;</span><br><span class="line">       <span class="built_in">return</span> response.getData();</span><br><span class="line">   &#125;</span><br><span class="line">.....</span><br></pre></td></tr></table></figure>
<p>可以看见都是先进行下路径的校验，然后生成一个RequestHeader,为请求头设置操作的类型<br>对于有接受stat参数的，最后会调用DataTree.copyStat(response.getStat(), stat)把最新的stat数据复制给传入的stat对象<br>根据类型生成不同的Record实现类，record接口提供两个方法，序列化与反序列化，一个Reqeust实现类实例以及一个Response实现类实例<br>当有回调函数时，将回调函数传入<br>最终都是调用ClientCnxn的submitRequest方法，获得ReplyHeader（实际的response数据会装载到传入的response对象里）<br>ReplyHeader只保存了是否出现失败，以及错误码和路径等信息，不保存实际返回的响应数据<br>submitRequest实际会调用queuePacket方法去发送给服务端，并获得及时响应的packet对象<br>因为是异步发送，后面会根据是否设置了超时时间选择调用wait阻塞等待或者等待超时设置replyHeader失败并返回<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br></pre></td><td class="code"><pre><span class="line">public ReplyHeader submitRequest(RequestHeader h, Record request,</span><br><span class="line">        Record response, WatchRegistration watchRegistration)</span><br><span class="line">        throws InterruptedException &#123;</span><br><span class="line">    <span class="built_in">return</span> submitRequest(h, request, response, watchRegistration, null);</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">public ReplyHeader submitRequest(RequestHeader h, Record request,</span><br><span class="line">        Record response, WatchRegistration watchRegistration,</span><br><span class="line">        WatchDeregistration watchDeregistration)</span><br><span class="line">        throws InterruptedException &#123;</span><br><span class="line">    ReplyHeader r = new ReplyHeader();</span><br><span class="line">    Packet packet = queuePacket(h, r, request, response, null, null, null,</span><br><span class="line">            null, watchRegistration, watchDeregistration);</span><br><span class="line">    synchronized (packet) &#123;</span><br><span class="line">        <span class="keyword">if</span> (requestTimeout &gt; 0) &#123;</span><br><span class="line">            // Wait <span class="keyword">for</span> request completion with timeout</span><br><span class="line">            waitForPacketFinish(r, packet);</span><br><span class="line">        &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">            // Wait <span class="keyword">for</span> request completion infinitely</span><br><span class="line">            <span class="keyword">while</span> (!packet.finished) &#123;</span><br><span class="line">                packet.wait();</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">if</span> (r.getErr() == Code.REQUESTTIMEOUT.intValue()) &#123;</span><br><span class="line">        sendThread.cleanAndNotifyState();</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="built_in">return</span> r;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure></p>
<p>queuePacket实际将请求信息封装为一个packet，并将packet加入outgoingQueue队列，由sendThread去处理，返回packet对象<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br></pre></td><td class="code"><pre><span class="line">private final LinkedBlockingDeque&lt;Packet&gt; outgoingQueue = new LinkedBlockingDeque&lt;Packet&gt;();</span><br><span class="line">public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,</span><br><span class="line">           Record response, AsyncCallback cb, String clientPath,</span><br><span class="line">           String serverPath, Object ctx, WatchRegistration watchRegistration,</span><br><span class="line">           WatchDeregistration watchDeregistration) &#123;</span><br><span class="line">       Packet packet = null;</span><br><span class="line"></span><br><span class="line">       // Note that we <span class="keyword">do</span> not generate the Xid <span class="keyword">for</span> the packet yet. It is</span><br><span class="line">       // generated later at send-time, by an implementation of ClientCnxnSocket::doIO(),</span><br><span class="line">       // <span class="built_in">where</span> the packet is actually sent.</span><br><span class="line">       packet = new Packet(h, r, request, response, watchRegistration);</span><br><span class="line">       packet.cb = cb;</span><br><span class="line">       packet.ctx = ctx;</span><br><span class="line">       packet.clientPath = clientPath;</span><br><span class="line">       packet.serverPath = serverPath;</span><br><span class="line">       packet.watchDeregistration = watchDeregistration;</span><br><span class="line">       // The synchronized block here is <span class="keyword">for</span> two purpose:</span><br><span class="line">       // 1. synchronize with the final cleanup() <span class="keyword">in</span> SendThread.run() to avoid race</span><br><span class="line">       // 2. synchronized against each packet. So <span class="keyword">if</span> a closeSession packet is added,</span><br><span class="line">       // later packet will be notified.</span><br><span class="line">       synchronized (state) &#123;</span><br><span class="line">           <span class="keyword">if</span> (!state.isAlive() || closing) &#123;</span><br><span class="line">               conLossPacket(packet);</span><br><span class="line">           &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">               // If the client is asking to close the session <span class="keyword">then</span></span><br><span class="line">               // mark as closing</span><br><span class="line">               <span class="keyword">if</span> (h.getType() == OpCode.closeSession) &#123;</span><br><span class="line">                   closing = <span class="literal">true</span>;</span><br><span class="line">               &#125;</span><br><span class="line">               outgoingQueue.add(packet);</span><br><span class="line">           &#125;</span><br><span class="line">       &#125;</span><br><span class="line">       sendThread.getClientCnxnSocket().packetAdded();</span><br><span class="line">       <span class="built_in">return</span> packet;</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure></p>
<p>processEvent方法根据event的类型，执行不同操作<br>如果是本地监听事件，直接调用注册的各个watcher的process方法<br>如果是本地的callBack回调函数类型，则使用instanceOf判断以后，使用不同的回调函数强制转换去执行processResult（从目前的源码分析暂时没看到有使用，可能是nio的socket实现类，这里只分析了默认的实现类ClientCnxnSocketNetty）<br>如果是response类型，使用instanceOf判断以后，选择具体的实现类去执行processResult操作<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br><span class="line">131</span><br><span class="line">132</span><br><span class="line">133</span><br><span class="line">134</span><br><span class="line">135</span><br><span class="line">136</span><br><span class="line">137</span><br><span class="line">138</span><br><span class="line">139</span><br><span class="line">140</span><br><span class="line">141</span><br><span class="line">142</span><br><span class="line">143</span><br><span class="line">144</span><br><span class="line">145</span><br><span class="line">146</span><br><span class="line">147</span><br><span class="line">148</span><br><span class="line">149</span><br><span class="line">150</span><br><span class="line">151</span><br><span class="line">152</span><br><span class="line">153</span><br><span class="line">154</span><br></pre></td><td class="code"><pre><span class="line">private void processEvent(Object event) &#123;</span><br><span class="line">  try &#123;</span><br><span class="line">	  <span class="keyword">if</span> (event instanceof WatcherSetEventPair) &#123;</span><br><span class="line">		  // each watcher will process the event</span><br><span class="line">		  WatcherSetEventPair pair = (WatcherSetEventPair) event;</span><br><span class="line">		  <span class="keyword">for</span> (Watcher watcher : pair.watchers) &#123;</span><br><span class="line">			  try &#123;</span><br><span class="line">				  watcher.process(pair.event);</span><br><span class="line">			  &#125; catch (Throwable t) &#123;</span><br><span class="line">				  LOG.error(<span class="string">"Error while calling watcher "</span>, t);</span><br><span class="line">			  &#125;</span><br><span class="line">		  &#125;</span><br><span class="line">		&#125; <span class="keyword">else</span> <span class="keyword">if</span> (event instanceof LocalCallback) &#123;</span><br><span class="line">			LocalCallback lcb = (LocalCallback) event;</span><br><span class="line">			<span class="keyword">if</span> (lcb.cb instanceof StatCallback) &#123;</span><br><span class="line">				((StatCallback) lcb.cb).processResult(lcb.rc, lcb.path,</span><br><span class="line">						lcb.ctx, null);</span><br><span class="line">			&#125; <span class="keyword">else</span> <span class="keyword">if</span> (lcb.cb instanceof DataCallback) &#123;</span><br><span class="line">				((DataCallback) lcb.cb).processResult(lcb.rc, lcb.path,</span><br><span class="line">						lcb.ctx, null, null);</span><br><span class="line">			&#125; <span class="keyword">else</span> <span class="keyword">if</span> (lcb.cb instanceof ACLCallback) &#123;</span><br><span class="line">				((ACLCallback) lcb.cb).processResult(lcb.rc, lcb.path,</span><br><span class="line">						lcb.ctx, null, null);</span><br><span class="line">			&#125; <span class="keyword">else</span> <span class="keyword">if</span> (lcb.cb instanceof ChildrenCallback) &#123;</span><br><span class="line">				((ChildrenCallback) lcb.cb).processResult(lcb.rc,</span><br><span class="line">						lcb.path, lcb.ctx, null);</span><br><span class="line">			&#125; <span class="keyword">else</span> <span class="keyword">if</span> (lcb.cb instanceof Children2Callback) &#123;</span><br><span class="line">				((Children2Callback) lcb.cb).processResult(lcb.rc,</span><br><span class="line">						lcb.path, lcb.ctx, null, null);</span><br><span class="line">			&#125; <span class="keyword">else</span> <span class="keyword">if</span> (lcb.cb instanceof StringCallback) &#123;</span><br><span class="line">				((StringCallback) lcb.cb).processResult(lcb.rc,</span><br><span class="line">						lcb.path, lcb.ctx, null);</span><br><span class="line">			&#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				((VoidCallback) lcb.cb).processResult(lcb.rc, lcb.path,</span><br><span class="line">						lcb.ctx);</span><br><span class="line">			&#125;</span><br><span class="line">		&#125; <span class="keyword">else</span> &#123;</span><br><span class="line">		  Packet p = (Packet) event;</span><br><span class="line">		  int rc = 0;</span><br><span class="line">		  String clientPath = p.clientPath;</span><br><span class="line">		  <span class="keyword">if</span> (p.replyHeader.getErr() != 0) &#123;</span><br><span class="line">			  rc = p.replyHeader.getErr();</span><br><span class="line">		  &#125;</span><br><span class="line">		  <span class="keyword">if</span> (p.cb == null) &#123;</span><br><span class="line">			  LOG.warn(<span class="string">"Somehow a null cb got to EventThread!"</span>);</span><br><span class="line">		  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof ExistsResponse</span><br><span class="line">				  || p.response instanceof SetDataResponse</span><br><span class="line">				  || p.response instanceof SetACLResponse) &#123;</span><br><span class="line">			  StatCallback cb = (StatCallback) p.cb;</span><br><span class="line">			  <span class="keyword">if</span> (rc == 0) &#123;</span><br><span class="line">				  <span class="keyword">if</span> (p.response instanceof ExistsResponse) &#123;</span><br><span class="line">					  cb.processResult(rc, clientPath, p.ctx,</span><br><span class="line">							  ((ExistsResponse) p.response)</span><br><span class="line">									  .getStat());</span><br><span class="line">				  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof SetDataResponse) &#123;</span><br><span class="line">					  cb.processResult(rc, clientPath, p.ctx,</span><br><span class="line">							  ((SetDataResponse) p.response)</span><br><span class="line">									  .getStat());</span><br><span class="line">				  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof SetACLResponse) &#123;</span><br><span class="line">					  cb.processResult(rc, clientPath, p.ctx,</span><br><span class="line">							  ((SetACLResponse) p.response)</span><br><span class="line">									  .getStat());</span><br><span class="line">				  &#125;</span><br><span class="line">			  &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, null);</span><br><span class="line">			  &#125;</span><br><span class="line">		  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof GetDataResponse) &#123;</span><br><span class="line">			  DataCallback cb = (DataCallback) p.cb;</span><br><span class="line">			  GetDataResponse rsp = (GetDataResponse) p.response;</span><br><span class="line">			  <span class="keyword">if</span> (rc == 0) &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, rsp</span><br><span class="line">						  .getData(), rsp.getStat());</span><br><span class="line">			  &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, null,</span><br><span class="line">						  null);</span><br><span class="line">			  &#125;</span><br><span class="line">		  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof GetACLResponse) &#123;</span><br><span class="line">			  ACLCallback cb = (ACLCallback) p.cb;</span><br><span class="line">			  GetACLResponse rsp = (GetACLResponse) p.response;</span><br><span class="line">			  <span class="keyword">if</span> (rc == 0) &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, rsp</span><br><span class="line">						  .getAcl(), rsp.getStat());</span><br><span class="line">			  &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, null,</span><br><span class="line">						  null);</span><br><span class="line">			  &#125;</span><br><span class="line">		  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof GetChildrenResponse) &#123;</span><br><span class="line">			  ChildrenCallback cb = (ChildrenCallback) p.cb;</span><br><span class="line">			  GetChildrenResponse rsp = (GetChildrenResponse) p.response;</span><br><span class="line">			  <span class="keyword">if</span> (rc == 0) &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, rsp</span><br><span class="line">						  .getChildren());</span><br><span class="line">			  &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, null);</span><br><span class="line">			  &#125;</span><br><span class="line">		  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof GetChildren2Response) &#123;</span><br><span class="line">			  Children2Callback cb = (Children2Callback) p.cb;</span><br><span class="line">			  GetChildren2Response rsp = (GetChildren2Response) p.response;</span><br><span class="line">			  <span class="keyword">if</span> (rc == 0) &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, rsp</span><br><span class="line">						  .getChildren(), rsp.getStat());</span><br><span class="line">			  &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, null, null);</span><br><span class="line">			  &#125;</span><br><span class="line">		  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof CreateResponse) &#123;</span><br><span class="line">			  StringCallback cb = (StringCallback) p.cb;</span><br><span class="line">			  CreateResponse rsp = (CreateResponse) p.response;</span><br><span class="line">			  <span class="keyword">if</span> (rc == 0) &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx,</span><br><span class="line">						  (chrootPath == null</span><br><span class="line">								  ? rsp.getPath()</span><br><span class="line">								  : rsp.getPath()</span><br><span class="line">							.substring(chrootPath.length())));</span><br><span class="line">			  &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, null);</span><br><span class="line">			  &#125;</span><br><span class="line">		  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof Create2Response) &#123;</span><br><span class="line">			  Create2Callback cb = (Create2Callback) p.cb;</span><br><span class="line">			  Create2Response rsp = (Create2Response) p.response;</span><br><span class="line">			  <span class="keyword">if</span> (rc == 0) &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx,</span><br><span class="line">						  (chrootPath == null</span><br><span class="line">								  ? rsp.getPath()</span><br><span class="line">								  : rsp.getPath()</span><br><span class="line">							.substring(chrootPath.length())), rsp.getStat());</span><br><span class="line">			  &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, null, null);</span><br><span class="line">			  &#125;                   </span><br><span class="line">		  &#125; <span class="keyword">else</span> <span class="keyword">if</span> (p.response instanceof MultiResponse) &#123;</span><br><span class="line">			  MultiCallback cb = (MultiCallback) p.cb;</span><br><span class="line">			  MultiResponse rsp = (MultiResponse) p.response;</span><br><span class="line">			  <span class="keyword">if</span> (rc == 0) &#123;</span><br><span class="line">				  List&lt;OpResult&gt; results = rsp.getResultList();</span><br><span class="line">				  int newRc = rc;</span><br><span class="line">				  <span class="keyword">for</span> (OpResult result : results) &#123;</span><br><span class="line">					  <span class="keyword">if</span> (result instanceof ErrorResult</span><br><span class="line">							  &amp;&amp; KeeperException.Code.OK.intValue() != (newRc = ((ErrorResult) result)</span><br><span class="line">							  .getErr())) &#123;</span><br><span class="line">						  <span class="built_in">break</span>;</span><br><span class="line">					  &#125;</span><br><span class="line">				  &#125;</span><br><span class="line">				  cb.processResult(newRc, clientPath, p.ctx, results);</span><br><span class="line">			  &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">				  cb.processResult(rc, clientPath, p.ctx, null);</span><br><span class="line">			  &#125;</span><br><span class="line">		  &#125;  <span class="keyword">else</span> <span class="keyword">if</span> (p.cb instanceof VoidCallback) &#123;</span><br><span class="line">			  VoidCallback cb = (VoidCallback) p.cb;</span><br><span class="line">			  cb.processResult(rc, clientPath, p.ctx);</span><br><span class="line">		  &#125;</span><br><span class="line">	  &#125;</span><br><span class="line">  &#125; catch (Throwable t) &#123;</span><br><span class="line">	  LOG.error(<span class="string">"Caught unexpected throwable"</span>, t);</span><br><span class="line">  &#125;</span><br><span class="line">  &#125;</span><br></pre></td></tr></table></figure></p>
<p>当在finishPacket执行完notify后，就会唤醒前面的对应函数，像是getData的，会判断返回头信息是否需要抛出异常<br>如果没有异常，则装载stat并调用response的getData获得服务端返回的数据并返回<br><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br></pre></td><td class="code"><pre><span class="line">ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);</span><br><span class="line"><span class="keyword">if</span> (r.getErr() != 0) &#123;</span><br><span class="line">	throw KeeperException.create(KeeperException.Code.get(r.getErr()),</span><br><span class="line">			clientPath);</span><br><span class="line">&#125;</span><br><span class="line"><span class="keyword">if</span> (<span class="built_in">stat</span> != null) &#123;</span><br><span class="line">	DataTree.copyStat(response.getStat(), <span class="built_in">stat</span>);</span><br><span class="line">&#125;</span><br><span class="line"><span class="built_in">return</span> response.getData();</span><br></pre></td></tr></table></figure></p>
<p>吐槽一下，zk里面使用netty做异步io，源码跳来跳去的，这博客1137行（–）</p>
<h3 id="更多精彩内容"><a href="#更多精彩内容" class="headerlink" title="更多精彩内容"></a>更多精彩内容</h3><h3 id="请移步"><a href="#请移步" class="headerlink" title="请移步"></a>请移步</h3><p>个人主页: <a href="http://yangyitao.top" target="_blank" rel="noopener">yangyitao.top</a></p>

      
    </div>
    
  </div>
  
    
<nav id="article-nav">
  
    <a href="/blog/2019/08/01/基于zookeeper实现分布式锁/" id="article-nav-newer" class="article-nav-link-wrap">
      <strong class="article-nav-caption">&lt;</strong>
      <div class="article-nav-title">
        
          基于zookeeper实现分布式锁
        
      </div>
    </a>
  
  
    <a href="/blog/2019/07/28/zookeeper源码分析-选举算法/" id="article-nav-older" class="article-nav-link-wrap">
      <div class="article-nav-title">zookeeper源码分析-选举算法</div>
      <strong class="article-nav-caption">&gt;</strong>
    </a>
  
</nav>

  
</article>


<div class="ds-share share" data-thread-key="zookeeper源码分析-事件注册及通信原理" data-title="zookeeper源码分析-事件注册及通信原理" data-url="http://eetal.gitee.io/blog/blog/2019/07/31/zookeeper源码分析-事件注册及通信原理/"  data-images="/blog/images/head.jpg" data-content="zookeeper源码分析-事件注册及通信原理">
    <div class="ds-share-inline">
      <ul  class="ds-share-icons-16">
      	<li data-toggle="ds-share-icons-more"><a class="ds-more" href="javascript:void(0);">分享到：</a></li>
        <li><a class="ds-weibo" href="javascript:void(0);" data-service="weibo">微博</a></li>
        <li><a class="ds-qzone" href="javascript:void(0);" data-service="qzone">QQ空间</a></li>
        <li><a class="ds-qqt" href="javascript:void(0);" data-service="qqt">腾讯微博</a></li>
        <li><a class="ds-wechat" href="javascript:void(0);" data-service="wechat">微信</a></li>
      </ul>
      <div class="ds-share-icons-more">
      </div>
    </div>
 </div>
 





</div>
      <footer id="footer">
  <div class="outer">
    <div id="footer-info">
      <div class="footer-left">
        &copy; 2019 Eetal
      </div>
        <div class="footer-right">
          <a href="http://hexo.io/" target="_blank">Hexo</a>  Theme <a href="https://github.com/smackgg/hexo-theme-smackdown" target="_blank">Smackdown</a>
        </div>
    </div>
  </div>
</footer>
    </div>
    
  <link rel="stylesheet" href="/blog/fancybox/jquery.fancybox.css">


<script>
	var yiliaConfig = {
		fancybox: true,
		mathjax: true,
		animate: true,
		isHome: false,
		isPost: true,
		isArchive: false,
		isTag: false,
		isCategory: false,
		open_in_new: false
	}
</script>
<script src="/blog/js/main.js"></script>



<script type="text/x-mathjax-config">
MathJax.Hub.Config({
    tex2jax: {
        inlineMath: [ ['$','$'], ["\\(","\\)"]  ],
        processEscapes: true,
        skipTags: ['script', 'noscript', 'style', 'textarea', 'pre', 'code']
    }
});

MathJax.Hub.Queue(function() {
    var all = MathJax.Hub.getAllJax(), i;
    for(i=0; i < all.length; i += 1) {
        all[i].SourceElement().parentNode.className += ' has-jax';                 
    }       
});
</script>

<script src="//cdn.bootcss.com/mathjax/2.7.0/MathJax.js"></script>


  </div>
</body>
</html>